package cn._51doit.flink.day05;

import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;

/**
 * 所有的window，按照是否key后再划分窗口，分为KeyedWindow和NonKeyedWindow
 *
 * 如果没有keyBy就划分窗口，就是NonKeyedWindow，底层调用的是windowAll方法
 *
 * 如果是NonKeyedWindow，window和window operator对应的Task并行度永远为1
 *
 */
public class CountWindowAllDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        SingleOutputStreamOperator<Integer> nums = lines.map(Integer::parseInt);
        //没有keyBy，调用countWindowAll，按照条数划分窗口，当窗口中的数据达到一定的条数再输出计算结果
        AllWindowedStream<Integer, GlobalWindow> windowedStream = nums.countWindowAll(5);
        //划分窗口后，需要调用相应的方法，window operator
        SingleOutputStreamOperator<Integer> res = windowedStream.sum(0);
        res.print();
        env.execute();


    }
}
